package com.hzgj.bcl.soa.interceptor;

import com.hzgj.bcl.soa.Event;
import com.hzgj.bcl.soa.Interceptor;
import com.hzgj.bcl.soa.api.IdempotentException;
import com.hzgj.bcl.soa.api.IdempotentProcessor;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.rpc.message.Payload;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.util.lang.StrKit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * Created by warning on 2016/10/26.
 */
public class IdempotentInterceptor extends Interceptor {


    Map<String, Map<String, IdempotentProcessor>> idempotentProcessors;
    private Logger logger = LoggerFactory.getLogger(getClass());

    public IdempotentInterceptor(Map<String, Map<String, IdempotentProcessor>> idempotentProcessors) {
        this.idempotentProcessors = idempotentProcessors;
    }

    @Override
    public Object up(Event evt) {
        Message msg = (Message) evt.getArg();
        Object object = msg.getPayload();
        if (!(object instanceof Payload)) {
            return up_int.up(evt);
        }
        Payload payload = (Payload) object;
        if (StrKit.notBlank(payload.getBizId())) {
            String service = payload.getService();
            Map<String, IdempotentProcessor> methodIdempotentProcessor = idempotentProcessors.get(service);
            if (methodIdempotentProcessor != null) {
                IdempotentProcessor idempotentProcessor = methodIdempotentProcessor.get(payload.getMethod());
                if (idempotentProcessor != null) {
                    try {
                        if (idempotentProcessor.check(payload.getBizId())) {
                            Object reply = idempotentProcessor.get(payload.getBizId());
                            if (reply != null) {
                                sendReply((Long) msg.getHeader(Constants.MESSAGE_SID), reply, false);
                            } else {
                                sendReply((Long) msg.getHeader(Constants.MESSAGE_SID), new IdempotentException("duplicate bizId  " + payload.getBizId()), true);
                            }
                        } else {
                            return up_int.up(evt);
                        }
                    } catch (IdempotentException e) {
                        logger.error("{}", e);
                        sendReply((Long) msg.getHeader(Constants.MESSAGE_SID), e, true);
                    }
                } else {
                    sendReply((Long) msg.getHeader(Constants.MESSAGE_SID), new IdempotentException("can't find  idempotentProcessor by  " + payload.getMethod()), true);
                }
            } else {
                sendReply((Long) msg.getHeader(Constants.MESSAGE_SID), new IdempotentException("can't find  methodIdempotentProcessor by  " + service), true);
            }
        } else {
            return up_int.up(evt);
        }
        return null;
    }

    protected void sendReply(final Long req_id, Object reply, boolean is_exception) {
        Message rsp = new Message();
        rsp.putHeader(Constants.MESSAGE_TYPE, is_exception ? Message.EXC_RSP : Message.RSP);
        rsp.putHeader(Constants.MESSAGE_SID, req_id);
        rsp.setPayload(reply);
        down_int.down(new Event(Event.MSG, rsp));
    }

}
